package com.tengju.data.infrastructure.kafka.config;


import com.tengju.data.domain.model.alert.AlertEvent;
import com.tengju.data.domain.model.alert.AlertService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.listener.ConsumerAwareListenerErrorHandler;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.MessageHeaders;

@Slf4j
@Configuration
public class CustomListenerErrorHandler {

    @Autowired
    AlertService alertService;

    @Value("${binlog.error.sleep.time}")
    Long binlogErrorSleepTime;

    @Bean(name = "binlogConsumerListenerErrorHandler")
    public ConsumerAwareListenerErrorHandler binlogConsumerListenerErrorHandler() {
        return (message, e, consumer) -> {
            log.error("errorHandler", e);
            AlertEvent event = new AlertEvent(e.getMessage());
            StringBuilder causeErrorMsg = new StringBuilder();
            Throwable causeEx = e.getCause();
            if (causeEx != null) {
                causeErrorMsg.append(causeEx.getClass()).append(": ").append(causeEx.getMessage()).append("\n");
                StackTraceElement[] elements = causeEx.getStackTrace();
                int i = 0;
                for (StackTraceElement stackTraceElement : elements) {
                    causeErrorMsg.append("\t").append(stackTraceElement.toString()).append("\n");
                    if (++i == 5) {
                        break;
                    }
                }
            }
            event.addItem("异常信息", causeErrorMsg.toString());
            alertService.alert(event);
            MessageHeaders headers = message.getHeaders();
            consumer.seek(new org.apache.kafka.common.TopicPartition(
                            headers.get(KafkaHeaders.RECEIVED_TOPIC, String.class),
                            headers.get(KafkaHeaders.RECEIVED_PARTITION_ID, Integer.class)),
                    headers.get(KafkaHeaders.OFFSET, Long.class));
            try {
                Thread.sleep(binlogErrorSleepTime);
            } catch (Exception exception) {
                log.error("thread sleep error", exception);
            }
            return null;
        };
    }
}